/**
 * Alipay.com Inc.
 * Copyright (c) 2004-2020 All Rights Reserved.
 */
package com.dal.flink.udf;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/common.html
 *
 * @author dingaolin
 * @version : TableSqlJob.java, v 0.1 2020年10月10日 下午2:32 dingaolin Exp $
 */
public class UdfSqlJob {

    public static void main(String[] args) throws Exception {
        // Create a TableEnvironment
        StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        bsEnv.setParallelism(1);
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);

        // mock data
        DataStream<String> teacherTextStream = bsEnv.socketTextStream("localhost", 9001);
        DataStream<Teacher> teacherDataStream = teacherTextStream.map((MapFunction<String, Teacher>) line -> {
            String[] split = line.split(",");
            return new Teacher(split[0], split[1]);
        });

        // Convert a DataStream or DataSet into a Table
        Table teacher = tableEnv.fromDataStream(teacherDataStream, $("uid"), $("name"));

        // Create Tables in the Catalog：Create a Table
        tableEnv.createTemporaryView("teacher", teacher);

        // register function
        tableEnv.createTemporarySystemFunction("subUdf", SubstringFunction.class);

        // Query a Table：SQL
        Table teacherResult = tableEnv.sqlQuery("SELECT uid, subUdf(name), subUdf(uid, name) FROM teacher");

        // Emit a Table
        DataStream<Row> teacherSinkStream = tableEnv.toAppendStream(teacherResult, Row.class);
        teacherSinkStream.print();

        // execute
        bsEnv.execute("TableSqlJob");
    }

    public static class Teacher {
        public String uid;
        public String name;

        //需要无参数构造方法
        public Teacher() {
        }

        public Teacher(String uid, String name) {
            this.uid = uid;
            this.name = name;
        }
    }
}